package com.atguigu.dataStreamApi.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Date;
import java.util.UUID;

public class Flink04_DataGenSource {
    public static void main(String[] args) {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);

            //DataGeneratorSource：数据生成器源
        DataGeneratorSource<String> generatorSource
                = new DataGeneratorSource<String>(
                new GeneratorFunction<Long, String>() {
                    @Override
                    public String map(Long value) throws Exception {
                        return UUID.randomUUID().toString() + "->" + value;//随机返回一个字符串,UUID是随机码，value是值
                    }
                },100 //数据生成条数
                 ,RateLimiterStrategy.perSecond(1) //rate：速率，limiter：限制，strategy：策略
                 ,Types.STRING //生成数据类型
        );
        DataStreamSource<String> ds
                = env.fromSource(generatorSource, WatermarkStrategy.noWatermarks(), "DatageneratorSource");

        ds.print();

        try {
                env.execute();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
}
